-module(rtps_sedp).

%%% @doc The `rtps_sedp' module implements the Endpoint Discovery
%%% Protocol, which defines the required information exchange between
%%% two Participants in order to discover each other’s Writer and
%%% Reader Endpoints.

-behaviour(gen_statem).

%% -include_lib("kernel/include/logger.hrl").
-include("rtps.hrl").
-include("rtps_spdp.hrl").
-include("rtps_sedp.hrl").
-include("rtps_history_cache.hrl").

%% API
-export([start_link/2,
	 setup/2, add/2, update/2, remove/2,
	 insert/3, modify/3, delete/2]).

%% gen_statem callbacks
-export([init/1, callback_mode/0,
	 idle/3,
	 running/3, 
	 handle_event/4, 
	 terminate/3, code_change/4]).

-export_type([writer_proxy/0, publication_builtin_topic_data/0,
	      reader_proxy/0, subscription_builtin_topic_data/0]).
-type writer_proxy() :: #writer_proxy{}.
-type publication_builtin_topic_data() :: #publication_builtin_topic_data{}.
-type reader_proxy() :: #reader_proxy{}.
-type subscription_builtin_topic_data() :: #subscription_builtin_topic_data{}.

-record(discovered_reader_data,
	{reader_proxy :: #reader_proxy{},
	 subscription_builtin_topic_data :: #subscription_builtin_topic_data{},
	 contentFilter :: term()}).

-record(discovered_writer_data, 
	{writer_proxy :: #writer_proxy{},
	 publication_builtin_topic_data :: #publication_builtin_topic_data{}}).

-record(discovered_topic_data, 
	{topic_builtin_topic_data :: #topic_builtin_topic_data{}}).

-record(data, {sup :: pid(), opts = [] :: [proplists:property()], cache :: pid(),
	       reg :: pid() | undefined, endpoints_sup :: pid() | undefined,
	       writers :: [{publications | subscriptions | topics, pid(), rtps:entity_id()}],
	       readers :: [{publications | subscriptions | topics, pid(), rtps:entity_id()}]
	      }).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Sup, Opts) ->
    gen_statem:start_link(?MODULE, [Sup, Opts], []).

-spec setup(Pid, Builtin_endpoints) -> ok when
      Pid :: pid(),
      Builtin_endpoints :: [rtps:available_builtin_endpoints()]. 
%% @doc Setup the SEDP. The participant uses this function to
%% configure the SEDP with the builtin endpoints required. Which
%% endpoints are used by the SEDP depends on the configuration of the
%% participant as determined by the user application.
setup(Pid, Builtin_endpoints) ->
    gen_statem:call(Pid, {setup, Builtin_endpoints}).

-spec add(Pid, Participant_proxy) -> ok when
      Pid :: pid(),
      Participant_proxy :: rtps_spdp:participant_proxy().
%% @doc This operation adds a remote participant to the SEDP. On
%% detecting a new remote participant, this function is called by the
%% SPDP after which the SEDP will create the proxies for the builtin
%% endpoints required to discover the remote participant's endpoints.
add(Pid, Participant_proxy) ->
    gen_statem:call(Pid, {add, Participant_proxy}).

-spec update(Pid, Participant_proxy) -> ok when
      Pid :: pid(),
      Participant_proxy :: rtps_spdp:participant_proxy().
%% @doc This operation informs the SEDP that an attribute of a remote
%% participant has changed which may be of interest to the SEDP.
%% TODO: implement this, but what can be updated?
update(_Pid, _Participant_proxy) ->
    error(not_implemented).
    %% gen_statem:call(Pid, {update, Participant_proxy}).

-spec remove(Pid, Guid_prefix) -> ok when
      Pid :: pid(),
      Guid_prefix :: rtps_spdp:guid_prefix().
%% @doc This operation is used to temove a remote participant from the
%% SEDP. This function is called by the SPDP, or maybe some other
%% process, when the SPDP concludes the remote participant isn't alive
%% anymore.
remove(Pid, Guid_prefix) ->
    gen_statem:cast(Pid, {remove, Guid_prefix}).

-spec insert(Pid, Proxy, Qos) -> ok when
      Pid :: pid(),
      Proxy :: writer_proxy() | reader_proxy(),
      Qos :: [rtps:parameter()].
%% @doc THis operation adds a local endpoint to the SEDP. When a new
%% endpoint is created within the local endpoint, the endpoint will
%% register itself with the SEDP, which will announce its availability
%% for remote SEDPs. NB: only endpoints with kind `user' are
%% registered.
insert(Pid, Proxy, Qos) ->
    gen_statem:call(Pid, {insert, Proxy, Qos}).

-spec modify(Pid, Proxy, Qos) -> ok when
      Pid :: pid(),
      Proxy :: writer_proxy() | reader_proxy(),
      Qos :: [rtps:parameter()].
%% @doc This operation is used for modifying some attribute of a registered endpoint.
modify(Pid, Proxy, Qos) ->
    gen_statem:call(Pid, {modify, Proxy, Qos}).
    
-spec delete(Pid, Proxy) -> ok when
      Pid :: pid(),
      Proxy :: writer_proxy() | reader_proxy().
%% @doc Delete the registration of a local endpoint from the SEDP.
delete(Pid, Proxy) ->
    gen_statem:cast(Pid, {delete, Proxy}).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @private
init([Sup, Opts]) ->
    Opts1 = [{history_kind, keep_all}, {reliability_level, reliable},
	     {history_durability, volatile}],
    Cache = rtps_history_cache:new(Opts1),
    {ok, idle, #data{sup = Sup, opts = Opts, cache = Cache, readers = [], writers = []}}.

%% @private
callback_mode() ->
    [state_functions, state_enter].

%% @private
idle(enter, _Old_state, #data{sup = Sup} = Data) ->
    Children = supervisor:which_children(Sup),
    {ok, Reg} = child(Children, registry),
    {ok, Endpoints_sup} = child(Children, endpoints),
    {keep_state, Data#data{reg = Reg, endpoints_sup = Endpoints_sup}};
idle({call, From}, {setup, Builtin_endpoints},
     #data{cache = Cache, reg = Reg, endpoints_sup = Endpoints_sup,
	   opts = Opts} = Data) ->
    Announcers = lists:filtermap(fun(publications_writer) -> {true, publications};
				    (subscriptions_writer) -> {true, subscriptions};
				    (topics_writer) -> {true, topics};
				    (_) -> false
			   end, Builtin_endpoints),
    Detectors = lists:filtermap(fun(publications_reader) -> {true, publications};
				   (subscriptions_reader) -> {true, subscriptions};
				   (topics_reader) -> {true, topics};
				   (_) -> false
			   end, Builtin_endpoints),
    Opts1 = [{state_type, stateful}, {reliability_level, reliable}, {history_cache, Cache},
	     {expects_inline_qos, false}, {topic_kind, with_key} | Opts],
    Fun1 = fun(Announcer, Acc) ->
		   {Entity_id_local, Entity_id_remote} = builtins(Announcer),
		   add_endpoint(Endpoints_sup, Reg, writer, Entity_id_local, Opts1),
		   Pid = rtps_reg:whereis_name(Reg, Entity_id_local),
		   [{Announcer, Pid, Entity_id_remote} | Acc]
	   end,
    Writers = lists:foldl(Fun1, [], Announcers),
    Fun2 = fun(Detector, Acc) ->
		   {Entity_id_remote, Entity_id_local} = builtins(Detector),
		   add_endpoint(Endpoints_sup, Reg, reader, Entity_id_local, Opts1),
		   Pid = rtps_reg:whereis_name(Reg, Entity_id_local),
		   [{Detector, Pid, Entity_id_remote} | Acc]
	   end,
    Readers = lists:foldl(Fun2, [], Detectors),

    %% Install event handler in the history cache's event manager
    ok = rtps_history_cache:add_sup_handler(Cache, rtps_sedp_eh, []),

    {next_state, running, Data#data{writers = Writers, readers = Readers},
     [{reply, From, ok}]}.

%% @private
running({call, From}, {add, #participant_proxy{guid_prefix = Guid_prefix_remote, 
					       metatraffic_unicast_locator_list = Meta_u_locs,
					       metatraffic_multicast_locator_list = Meta_m_locs}},
    	#data{writers = Writers, readers = Readers}) ->
    Guid = #guid{guid_prefix = Guid_prefix_remote},
    Opts1 = [{multicast_locator_list, Meta_m_locs}],
    Opts2 = case Meta_u_locs of
    		undefined -> Opts1;
    		_ -> [{unicast_locator_list, Meta_u_locs} | Opts1]
    	    end,
    Fun1 = fun({_Announcer, Pid, Entity_id_remote}) ->
    		   Guid_remote = Guid#guid{entity_id = Entity_id_remote},
    		   rtps_writer:matched_reader_add(Pid, Guid_remote, Opts2)
    	   end,
    lists:foreach(Fun1, Writers),
    Fun2 = fun({_Detector, Pid, Entity_id_remote}) ->
    		   Guid_remote = Guid#guid{entity_id = Entity_id_remote},
    		   rtps_reader:matched_writer_add(Pid, Guid_remote, Opts2)
    	   end,
    lists:foreach(Fun2, Readers),
    {keep_state_and_data, [{reply, From, ok}]};
running({call, From}, {update, #participant_proxy{guid_prefix = _Guid_prefix}},
    	#data{writers = _Writers, readers = _Readers}) ->
    Reply = {error, not_implemented},
    {keep_state_and_data, [{reply, From, Reply}]};
running(cast, {remove, Guid_prefix}, #data{writers = Writers, readers = Readers}) ->
    Fun1 = fun({_Announcer, Pid, Entity_id}) ->
    		   Guid = #guid{guid_prefix = Guid_prefix, entity_id = Entity_id},
    		   rtps_writer:matched_reader_remove(Pid, Guid)
    	   end,
    lists:foreach(Fun1, Writers),
    Fun2 = fun({_Detector, Pid, Entity_id}) ->
    		   Guid = #guid{guid_prefix = Guid_prefix, entity_id = Entity_id},
    		   rtps_reader:matched_writer_remove(Pid, Guid)
    	   end,
    lists:foreach(Fun2, Readers),
    keep_state_and_data;
running({call, From}, {insert, #writer_proxy{} = Writer_proxy, Qos},
	#data{writers = Writers} = Data) ->
    DWD = #discovered_writer_data{writer_proxy = Writer_proxy,
				  publication_builtin_topic_data = Qos},
    Reply = do_insert(publications, DWD, Writers),
    {keep_state, Data, [{reply, From, Reply}]};
running({call, From}, {insert, #reader_proxy{} = Reader_proxy, Qos},
	#data{writers = Writers} = Data) ->
    DRD = #discovered_reader_data{reader_proxy = Reader_proxy,
				  subscription_builtin_topic_data = Qos},
    Reply = do_insert(subscriptions, DRD, Writers),
    {keep_state, Data, [{reply, From, Reply}]};
running({call, From}, {modify, _Proxy}, _Data) ->
    Reply = {error, not_implemented},
    {keep_state_and_data, [{reply, From, Reply}]};
running(cast, {delete, #writer_proxy{remote_writer_guid = Guid}},
	#data{writers = Writers} = Data) ->
    do_delete(publications, Guid, Writers),
    {keep_state, Data};
running(cast, {delete, #reader_proxy{remote_reader_guid = Guid}},
	#data{writers = Writers} = Data) ->
    do_delete(subscriptions, Guid, Writers),
    {keep_state, Data};
running(Event_type, Event_content, Data) ->
    handle_event(Event_type, Event_content, running, Data).

%% @private
%% TODO: remove catch-all clauses
handle_event({call, From} = _Event_type, _Event_content, _State_name, _Data) ->
    Reply = ok,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event(_Event_type,  _Event_content, _State_name, _Data) ->
    keep_state_and_data.

%% @private
terminate(_Reason, _State_name, _Data) ->
    ok.

%% @private
code_change(_Old_vsn, State_name, Data, _Extra) ->
    {ok, State_name, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
child(Children, Id) ->
    case lists:keyfind(Id, 1, Children) of
	{_Id, Child, _Type, _Modules} when is_pid(Child) ->
	    {ok, Child};
	{_Id, Child, _Type, _Modules} when Child =:= restarting ->
	    {error, Child};
	_ ->
	    {error, notfound}
    end.

add_endpoint(Endpoints_sup, Reg, Kind, Entity_id, Opts) ->
    Spec = #{id => Entity_id,
	     start => {rtps_endpoint_sup, start_link, [Kind, Entity_id, Reg, Opts]},
	     restart => permanent,
	     type => supervisor},
    {ok, _} = rtps_ofo_sup:start_child(Endpoints_sup, Spec).

%% @private
builtins(publications) ->
    {'SEDPbuiltinPublicationsWriter', 'SEDPbuiltinPublicationsReader'};
builtins(subscriptions) ->
    {'SEDPbuiltinSubscriptionsWriter', 'SEDPbuiltinSubscriptionsReader'};
builtins(topics) ->
    {'SEDPbuiltinTopicsWriter', 'SEDPbuiltinTopicsReader'}.


%% @private
enc_ded(#discovered_reader_data{reader_proxy = RP, subscription_builtin_topic_data = _SBTD}) ->
    %% TODO: also encode SBTD. rtps_psm:enc_data(enc_rp(RP) ++ SBTD);
    rtps_psm:enc_data(enc_rp(RP));
enc_ded(#discovered_writer_data{writer_proxy = WP, publication_builtin_topic_data = _SBTD}) ->
    %% TODO: also encode SBTD. rtps_psm:enc_data(enc_wp(WP) ++ _SBTD).
    rtps_psm:enc_data(enc_wp(WP)).

%% @private
enc_rp(#reader_proxy{remote_reader_guid = Guid,
		     expects_inline_qos = Expects_inline_qos,
		     unicast_locator_list = U_locs,
		     multicast_locator_list = M_locs}) ->
    [#parameter{id = participant_guid, value = Guid},
     #parameter{id = expects_inline_qos, value = Expects_inline_qos}] ++
	enc_locs(uni, U_locs) ++
	enc_locs(multi, M_locs).
    
%% @private
enc_wp(#writer_proxy{remote_writer_guid = Guid,
		     unicast_locator_list = U_locs,
		     multicast_locator_list = M_locs}) ->
    [#parameter{id = participant_guid, value = Guid}] ++
	enc_locs(uni, U_locs) ++
	enc_locs(multi, M_locs).

%% @private
enc_locs(_, undefined) ->
    [];
enc_locs(uni, Locs) ->
    [#parameter{id = unicast_locator, value = Loc} || Loc <- Locs];
enc_locs(multi, Locs) ->
    [#parameter{id = multicast_locator, value = Loc} || Loc <- Locs].

%% @private
do_insert(Type, DD, Writers) ->
    case lists:keyfind(Type, 1, Writers) of
	{_Type, Pid, _Entity_id} ->
	    Serialized_data = enc_ded(DD),
	    {ok, Change} = rtps_writer:new_change(Pid, alive, Serialized_data, undefined),
	    rtps_writer:add_change(Pid, Change);
	false ->
	    ok
    end.

%% @private
do_delete(Type, Guid, Writers) ->
    case lists:keyfind(Type, 1, Writers) of
	{_Type, Pid, _Entity_id} ->
	    Instance_handle = rtps_psm:enc_guid(Guid),
	    {ok, Change} = rtps_writer:new_change(Pid, not_alive_disposed, undefined, Instance_handle),
	    rtps_writer:add_change(Pid, Change);
	_ ->
	    ok
    end.
    
